package com.example.webflux.test;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import java.io.IOException;
import java.lang.reflect.Array;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

public class ComplexReactorExample {

    public static void main(String[] args) {
        String[] arr = {"data1", "data2", "data3", "data4", "data5"};
        // 模拟一些数据源
        List<String> dataSources = Arrays.asList(arr);
        AtomicInteger idx = new AtomicInteger(0);

        // 创建 Flux 和进行操作
        Flux.fromIterable(dataSources)
            .doOnNext(data -> {
                // 处理每个数据
                String transformedData = transformData(data);
                try {
                    // 假设这里发送到某个异步系统
                    sendData(transformedData, idx);
                } catch (IOException e) {
                    // 错误处理
                    System.err.println("发送失败, 数据: " + transformedData + " 错误信息: " + e.getMessage());
                    // 重试机制
                    retrySend(transformedData, idx).subscribe();
                }
            })
            .doOnError(err -> {
                // 处理错误并记录
                System.err.println("遇到错误: " + err.getMessage());
            })
            .doOnComplete(() -> {
                // 完成后的回调
                System.out.println("处理完成！");
            })
            .buffer(3) // 每批处理 3 条数据
            .flatMap(batch -> {
                // 批处理后再异步处理
                return Mono.just(batch)
                        .flatMap(b -> processBatchAsync(b));
            })
            .subscribe(
                result -> System.out.println("异步处理结果: " + result),
                error -> System.err.println("最终错误: " + error.getMessage())
            );
    }

    // 模拟数据转换
    private static String transformData(String data) {
        return data.toUpperCase();
    }

    // 模拟数据发送
    private static void sendData(String data, AtomicInteger idx) throws IOException {
        if (Math.random() > 0.7) { // 随机失败
            throw new IOException("发送失败");
        }
        System.out.println("发送数据成功, 数据: " + data + ", ID: " + idx.incrementAndGet());
    }

    // 重试机制
    private static Mono<Object> retrySend(String data, AtomicInteger idx) {
        return Mono.defer(() -> {
            try {
                sendData(data, idx);
                return Mono.empty();
            } catch (IOException e) {
                return Mono.error(e);
            }
        }).retry(3)  // 最大重试次数 3 次
          .doOnTerminate(() -> System.out.println("重试结束"));
    }

    // 模拟异步批量处理
    private static Mono<String> processBatchAsync(List<String> batch) {
        return Mono.fromCallable(() -> {
            // 模拟一些耗时操作
            Thread.sleep(1000);
            return batch.stream().collect(Collectors.joining(", "));
        });
    }
}
